/**
 * 
 */
package org.demetra.core;

import java.io.File;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Random;

import org.demetra.core.JetsException;
import org.demetra.core.Server;
import org.demetra.core.WorkerLoop;
import org.demetra.jets.bdb.OtpErlangTupleSecondaryKeyCreator;

import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangBinary;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangString;
import com.ericsson.otp.erlang.OtpErlangTuple;
import com.ericsson.otp.erlang.OtpMbox;
import com.ericsson.otp.erlang.OtpNode;
import com.ericsson.otp.erlang.OtpOutputStream;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.collections.StoredSortedMap;
import com.sleepycat.db.Database;
import com.sleepycat.db.Environment;
import com.sleepycat.db.SecondaryDatabase;

/**
 * @author vvorobyov
 *
 */
public class JetsService implements Server {

	private static final String CLASS_CATALOG = "java_class_catalog";
	
	private static final String QUAD_STORE = "quad_store";
	private static final String INDEX_PREFIX = "index_";
	
	private File _dataRoot;
	private Environment _env;

	private StoredClassCatalog _javaCatalog;

	private Database _quadDb;
	
	private ArrayList<SecondaryDatabase> _indexes = new ArrayList<SecondaryDatabase>();

	private SerialBinding<OtpErlangTuple> _tupleBinding;

	private SerialBinding<OtpErlangObject> _tupleKeyBinding;

	private StoredMap<OtpErlangObject, OtpErlangTuple> _tupleMapView;

	private HashMap<Integer, StoredMap<OtpErlangObject, OtpErlangTuple>> _indexViews;

	/**
	 * 
	 */
	public JetsService(File aDataRoot) {
		_dataRoot = aDataRoot;
	}

	/* (non-Javadoc)
	 * @see org.demetra.core.Server#createSession(com.ericsson.otp.erlang.OtpMbox)
	 */
	@Override
	public WorkerLoop createSession(OtpMbox mbox) {
		return new JetsSession(mbox);
	}

	/* (non-Javadoc)
	 * @see org.demetra.core.Server#init(com.ericsson.otp.erlang.OtpNode)
	 */
	@Override
	public void init(OtpNode node) throws Exception {
		//Environment init
		EnvironmentConfig envConfig = new EnvironmentConfig();
//        envConfig.setTransactional(true);
		int Mb=1024*1024;
		envConfig.setCachePercent(80);
//		envConfig.setCacheSize(1200*Mb);
        envConfig.setAllowCreate(true);
//        envConfig.setConfigParam("je.evictor.lruOnly", "false");
//        envConfig.setConfigParam("je.evictor.nodesPerScan", "100");
        

        _env = new Environment(_dataRoot, envConfig);
        
        //Ctalog init
        DatabaseConfig dbConfig = new DatabaseConfig();
//        dbConfig.setTransactional(true);
        dbConfig.setAllowCreate(true);
        dbConfig.setDeferredWrite(true);
        dbConfig.setSortedDuplicates(false);
//        dbConfig.setNodeMaxEntries(1024);
        
        System.out.println(dbConfig.getNodeMaxDupTreeEntries());
        System.out.println(dbConfig.getNodeMaxEntries());

        Database catalogDb = _env.openDatabase(null, CLASS_CATALOG, dbConfig);
        
        _javaCatalog = new StoredClassCatalog(catalogDb);
        
        _tupleBinding =
            new SerialBinding<OtpErlangTuple>(_javaCatalog, OtpErlangTuple.class);
        _tupleKeyBinding =
        	new SerialBinding<OtpErlangObject>(_javaCatalog, OtpErlangObject.class);
        
        //Quad db init 
        _quadDb = _env.openDatabase(null, QUAD_STORE, dbConfig);
        
        SecondaryConfig secConfig = new SecondaryConfig();
//        secConfig.setTransactional(true);
        secConfig.setAllowCreate(true);
        secConfig.setDeferredWrite(true);
        secConfig.setSortedDuplicates(true);
//        secConfig.setNodeMaxEntries(1024);
//        secConfig.set

        for(int i=0; i<4; i++)
        {
//        	if(i==1)
//        		continue;
	        secConfig.setKeyCreator(
	            new OtpErlangTupleSecondaryKeyCreator(_javaCatalog,i));
	        _indexes.add(_env.openSecondaryDatabase(null, 
	                INDEX_PREFIX+i,
	                _quadDb,
	                secConfig));
        }
//        secConfig.setKeyCreator(
//	            new OtpErlangTupleSecondaryKeyCreator(_javaCatalog,0));
//	        _indexes.add(_env.openSecondaryDatabase(null, 
//	                INDEX_PREFIX+0,
//	                _quadDb,
//	                secConfig));
//	    secConfig.setKeyCreator(
//	        		new OtpErlangTupleSecondaryKeyCreator(_javaCatalog,2));
//	        _indexes.add(_env.openSecondaryDatabase(null, 
//	        		INDEX_PREFIX+2,
//	        		_quadDb,
//	        		secConfig));
	}

	/* (non-Javadoc)
	 * @see org.demetra.core.Server#shutdown()
	 */
	@Override
	public void shutdown() throws Exception {
		for(SecondaryDatabase index:_indexes)
			index.close();
		_quadDb.close();
		_javaCatalog.close();
		_env.close();
	}
	public static final OtpErlangBinary md5OfOtpErlangObject(OtpErlangObject aObj)
	{
		byte[] binRep=new OtpOutputStream(aObj).toByteArray();
		MessageDigest md5;
		try {
			md5 = MessageDigest.getInstance("MD5");
		} catch (NoSuchAlgorithmException e) {
			throw new JetsException(e);
		}
		md5.update(binRep);
		return new OtpErlangBinary(md5.digest());
	}
///////////////////////////////////////////////////////////////////////
	public synchronized final StoredMap<OtpErlangObject, OtpErlangTuple> tuples(){
		if(_tupleMapView==null)
			_tupleMapView = 
            new StoredMap<OtpErlangObject, OtpErlangTuple>(
            		_quadDb,
            		_tupleKeyBinding,
            		_tupleBinding,
            		true);
		return _tupleMapView;
	}
	public synchronized final StoredMap<OtpErlangObject, OtpErlangTuple> index(int aNum)
	{
		_indexViews = new HashMap<Integer, StoredMap<OtpErlangObject,OtpErlangTuple>>();
		StoredMap<OtpErlangObject, OtpErlangTuple> indexView=_indexViews.get(aNum);
		if(indexView==null)
		{
			indexView = new StoredSortedMap<OtpErlangObject, OtpErlangTuple>(
					_indexes.get(aNum),
					_tupleKeyBinding,
					_tupleBinding,
					true);
			_indexViews.put(aNum, indexView);
		}
		return indexView;
	}
	public void addTuples(OtpErlangList aTuples)
	{
		StoredMap<OtpErlangObject, OtpErlangTuple> tuples = tuples();
		for(OtpErlangObject obj:aTuples.elements())
		{
			OtpErlangTuple tuple = (OtpErlangTuple)obj;
			OtpErlangObject key = md5OfOtpErlangObject(tuple);
			tuples.put(key, tuple);
		}
	}
	public Collection<OtpErlangTuple> lookupIndex(int aIndexNum, OtpErlangObject aSecKey)
	{
		StoredMap<OtpErlangObject, OtpErlangTuple> index = index(aIndexNum);
		return index.duplicates(aSecKey);
	}
///Test
	public static void main(String[] args) throws Exception {
		String path ="~/bdb";
		if(args.length>0)
			path = args[0];
		JetsService jetsService = new JetsService(new File(path));
		jetsService.init(null);
		try
		{
		addBulk(jetsService);
//		addFew(jetsService);
//			for(Object obj:jetsService.lookupIndex(1, atom("b")))
//				System.out.println(obj);
		}
		finally
		{
			jetsService.shutdown();
		}
	}
	
	public void sync() throws DatabaseException
	{
		_quadDb.sync();
		for(Database idx:_indexes)
		{
			idx.sync();
		}
		_env.cleanLog();
	}
	
	public void printStats() throws DatabaseException
	{
		//System.err.println(_env.getStats(null));
	}

private static void addBulk(JetsService jetsService) throws DatabaseException {
			String bigOtpString = "ioweoruhiuweohqfqocvwhoxifwixfiowuqrfgiuoqhgiuwhgioqwehgwmxwqiomfzmoziqweoihwqioegoiwqemgziwmqhiemziwqoegioweoqwieioweqmweiqmmiqwgmeiwqmoimwei";

	long start=System.currentTimeMillis();
	for(int i=0;i<1000000;i++)
	{
		if(i%10000 == 0 && i>0)
		{
			jetsService.sync();
			jetsService.printStats();
			long fin = System.currentTimeMillis();
			System.out.println("Tot "+i+" speed "+40000f/(((float)(fin-start))/1000f));
			start=fin;
		}
		Random rnd = new Random(System.currentTimeMillis());
		jetsService.addTuples(list(
				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(2000))),atom("graph"+i)),
				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred1")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(2000))),atom("graph"+i)),
				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred2")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(2000))),atom("graph"+i)),
				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred3")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(2000))),atom("graph"+i))
				));
//		jetsService.addTuples(list(
//				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(4))),atom("graph"+i)),
//				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred1")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(4))),atom("graph"+i)),
//				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred2")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(4))),atom("graph"+i)),
//				tuple(atom("subj"+i),tuple(atom("onto"),atom("pred3")),tuple(atom("onto"),new OtpErlangString(bigOtpString+rnd.nextInt(4))),atom("graph"+i))
//		));
	}
}

private static void addFew(JetsService jetsService) {
	jetsService.addTuples(list(
			tuple(atom("a"),atom("b"),atom("c"),atom("d")),
			tuple(atom("a1"),atom("b"),atom("c"),atom("d")),
			tuple(atom("a2"),atom("b"),atom("c"),atom("d"))
			));
}
	
	protected static final OtpErlangList list(OtpErlangObject... aObjs)
	{
		return new OtpErlangList(aObjs);
	}
	protected static final OtpErlangTuple tuple(OtpErlangObject... aObjs)
	{
		return new OtpErlangTuple(aObjs);
	}

	protected static final OtpErlangAtom atom(String aStr)
	{
		return new OtpErlangAtom(aStr);
	}
}
